package cn.monkey.io.transport.mq.redisson;

import cn.monkey.io.transport.mq.MqOutboundConfig;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Stream;

public class SimpleRedissonOutbound implements RedissonOutbound {

    private final MqOutboundConfig outboundConfig;

    private final AtomicBoolean closed;

    private final BlockingQueue<Object> cache;

    private final Collection<RTopic> senders;

    public SimpleRedissonOutbound(RedissonClient redissonClient,
                                  Codec codec,
                                  MqOutboundConfig outboundConfig) {
        this.outboundConfig = outboundConfig;
        this.cache = outboundConfig.getCacheSize() > 0 ? new LinkedBlockingQueue<>(outboundConfig.getCacheSize()) : null;
        this.senders = Stream.of(outboundConfig.getTopics())
                .map(topic -> redissonClient.getTopic(topic, codec))
                .toList();
        this.closed = new AtomicBoolean(false);
    }

    void validClosed() {
        if (this.closed.get()) {
            throw new UnsupportedOperationException("outbound is closed");
        }
    }

    @Override
    public Mono<Void> write(Object o) {
        validClosed();
        if (this.cache == null) {
            return this.writeAndFlush(o);
        }
        boolean offer = this.cache.offer(o);
        return offer ? Mono.empty() : Mono.error(new UnsupportedOperationException("cache is full"));
    }

    @Override
    public Mono<Void> writeAndFlush(Object o) {
        validClosed();
        CompletableFuture<?>[] array = this.senders.stream()
                .map(topic -> topic.publishAsync(o).toCompletableFuture())
                .toArray(CompletableFuture[]::new);
        return Mono.fromCompletionStage(CompletableFuture.allOf(array));
    }

    @Override
    public Mono<Void> flush() {
        validClosed();
        if (this.cache.isEmpty()) {
            return Mono.empty();
        }
        List<Object> objects = new ArrayList<>(this.cache.size());
        this.cache.drainTo(objects);
        List<Mono<Void>> monoList = objects.stream()
                .map(this::writeAndFlush)
                .toList();
        return Flux.merge(monoList).then();
    }


    @Override
    public Mono<Void> close() {
        this.closed.set(true);
        return Mono.empty();
    }

    @Override
    public MqOutboundConfig outboundConfig() {
        return this.outboundConfig;
    }
}
